/*
 * Copyright 2015 MingxingWang

 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at

 *     http://www.apache.org/licenses/LICENSE-2.0

 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package com.audaque.graphdb

import java.io.{DataInputStream, DataOutputStream}

import scala.collection.mutable

/**
 * Created by Ming on 2015/2/9.
 */
class SchemaFactory {
  private var nextId: Long = 0
  private val schemas = mutable.HashMap[String, Schema]()
  private val id2schema = mutable.HashMap[Long, Schema]()
  private val name2schema = mutable.HashMap[String, Schema]()

  private def cmp(field: (String, Type)): String = {
    field._1 + field._2.tid
  }

  private def genKey(fields: Array[(String, Type)]): String = {
    if (fields == null || fields.length == 0) {
      return ""
    }
    val key = new mutable.StringBuilder()
    for ((name, tp) <- fields.sortBy(cmp)) {
      key.append(name.toLowerCase).append(tp.tid)
    }
    key.toString()
  }

  def genSchema(name: String, fields: Array[(String, Type)]): Schema = {
    if (name != null && !name.isEmpty) {
      name2schema.get(name.toLowerCase) match {
        case Some(sch) => return sch
        case None =>
          val key = genKey(fields)
          schemas.get(key) match {
            case Some(sch) =>
              name2schema += (name.toLowerCase -> sch)
              sch
            case None =>
              val sch = new Schema(nextId, fields)
              schemas += (key -> sch)
              id2schema += (nextId -> sch)
              name2schema += (name.toLowerCase -> sch)
              nextId += 1
              sch
          }
      }
    } else {
      val key = genKey(fields)
      schemas.get(key) match {
        case Some(sch) => sch
        case None =>
          val sch = new Schema(nextId, fields)
          schemas += (key -> sch)
          id2schema += (nextId -> sch)
          nextId += 1
          sch
      }
    }
  }

  def getSchemaById(sid: Long): Option[Schema] = {
    id2schema.get(sid)
  }

  def getSchemaByName(name: String): Option[Schema] = {
    name2schema.get(name.toLowerCase)
  }

  def save(out: DataOutputStream) = {
    out.writeLong(nextId)
    out.writeInt(schemas.size)
    for ((keys, sch) <- schemas) {
      out.writeUTF(keys)
      out.writeLong(sch.id)
      out.writeInt(sch.size)
      for (i <- 0 until sch.fields.length) {
        out.writeUTF(sch.getFieldName(i))
        out.writeInt(sch.getFieldType(i).tid)
      }
    }
    out.writeInt(name2schema.size)
    for ((name, sch) <- name2schema) {
      out.writeUTF(name)
      out.writeLong(sch.id)
    }
  }

  def load(in: DataInputStream) {
    schemas.clear()
    id2schema.clear()
    nextId = in.readLong()
    val size = in.readInt()
    for (i <- 0 until size) {
      val schKeys = in.readUTF()
      val schId = in.readLong()
      val fieldSize = in.readInt()
      val fields = new Array[(String, Type)](fieldSize)
      for (j <- 0 until fieldSize) {
        fields(j) = (in.readUTF(), Type(in.readInt()))
      }
      val sch = new Schema(schId, fields)
      schemas += (schKeys -> sch)
      id2schema += (schId -> sch)
    }
    name2schema.clear()
    val nameSize = in.readInt()
    for (i <- 0 until nameSize) {
      val schName = in.readUTF()
      val schId = in.readLong()
      id2schema.get(schId) match {
        case Some(sch) => name2schema += (schName -> sch)
        case None =>
      }
    }
  }
}
